package cn.data.stream.config;

import com.alibaba.fastjson.JSON;
import cn.data.stream.annotation.Output;
import cn.data.stream.annotation.DataStream;
import cn.data.stream.config.parse.ComponentNode;
import cn.data.stream.config.parse.DataStreamConfigInfo;
import cn.data.stream.enums.StreamType;
import cn.data.stream.expection.DataStreamInitException;
import cn.data.stream.stream.DataStreamManager;
import cn.data.stream.stream.IComponent;
import com.google.common.collect.Sets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.core.env.PropertyResolver;

/**
 * @author shi hong yu
 */
@EnableConfigurationProperties
@Slf4j
@Import(DataStreamConfig.class)
public class DataStreamAutoConfiguration {

    private DataStreamConfig streamConfig;

    @Autowired
    private ApplicationContext context;

    @Resource
    protected PropertyResolver config;

    public DataStreamAutoConfiguration(DataStreamConfig streamConfig) {
        log.info("==========================================");
        log.info("data-stream 配置:{}", JSON.toJSONString(streamConfig.getDataStreamConfig()));
        log.info("==========================================");
        this.streamConfig = streamConfig;
    }

    /**
     * 加载stream配置 构建上下游关系 并初始化stream<br> 暂时实现方案：注解配置和xml配置不同时生效，优先加载xml配置，无xml配置时加载注解配置
     *
     * @return
     */
    @Bean
    public DataStreamManager getDataStreamManager() throws DataStreamInitException {
        log.debug("开始加载data-stream配置");
        DataStreamManager manager = new DataStreamManager();
        //暂时实现方案：注解配置和xml配置不同时生效，优先加载xml配置，无xml配置时加载注解配置
        if (streamConfig == null || streamConfig.getDataStreamConfig() == null) {
            loadAnnotationStreamConfig(manager);
        } else {
            loadOutStreamConfig(manager);
        }
        log.debug("data-stream配置加载完成");
        return manager;
    }

    private DataStreamManager loadAnnotationStreamConfig(DataStreamManager manager) throws DataStreamInitException {
        Map<String, Object> beansWithAnnotation = context.getBeansWithAnnotation(DataStream.class);
        log.info("data-stream 加载注解配置 {}", beansWithAnnotation.keySet());
        for (Entry<String, Object> ent : beansWithAnnotation.entrySet()) {
            String currentStreamName = ent.getKey();
            IComponent currentComp =null;
            if (manager.containsComponent(currentStreamName)) {
                log.debug("data-stream 已加载{} ",currentStreamName);
                currentComp = manager.getComponent(currentStreamName);
            }else{
                currentComp = (IComponent) ent.getValue();
                currentComp.setName(currentStreamName);
                log.debug("data-stream 加载 {} {}",currentStreamName,currentComp.getClass().getName());
                DataStream currentDsAnno = AnnotationUtils.findAnnotation(currentComp.getClass(), DataStream.class);
                StreamType currStreamType = currentDsAnno.type();
                manager.addComponent(currStreamType, currentComp);
            }

            DataStream currentDsAnno = AnnotationUtils.findAnnotation(currentComp.getClass(), DataStream.class);
            StreamType currStreamType = currentDsAnno.type();
            Set<String> outputSet = Sets.newLinkedHashSet();
            String[] outputs1 = currentDsAnno.outputs();

            if (outputs1 != null) {
                log.info("加载DataStream注解 output配置{} : {}",currentStreamName,Arrays.toString(outputs1));
               Collections.addAll(outputSet,outputs1);
            }
            log.info("data-stream初始化{} componentName:{} beanName:{}", currStreamType, currentStreamName, currentStreamName);
            Output output = AnnotationUtils.findAnnotation(currentComp.getClass(), Output.class);
            if (output != null && output.outputs().length > 0) {
                log.info("加载Output注解配置{} : {}",currentStreamName,Arrays.toString(output.outputs()));
                Collections.addAll(outputSet,output.outputs());
            }

            String outStreamStr = config.getProperty(currentStreamName + ".outputs");
            if(StringUtils.isNotBlank(outStreamStr)){
                log.info("加载配置文件中 {}.outputs={}",currentStreamName,outStreamStr);
                String[] split = outStreamStr.split(",");
                Collections.addAll(outputSet,split);
            }
            if(CollectionUtils.isEmpty(outputSet)){
                continue;
            }
            log.info("{} [{}]设置下游组件:{}", currentStreamName, currentComp,outputSet);
            List<IComponent> collect = new ArrayList<>();
            for (String o : outputSet) {
                String[] split = o.split(":");
                String nextStreamName = split[0];
                String nextBeanName = split.length < 2 ? split[0] : split[1];
                IComponent nextComp = null;
                if(manager.containsComponent(nextStreamName)){
                    nextComp = manager.getComponent(nextStreamName);
                    collect.add(nextComp);
                    log.info("data-stream {} 下游关系 {} --> {}", currStreamType, currentStreamName, nextStreamName);
                    continue;
                }

                if (!beansWithAnnotation.containsKey(nextBeanName)) {
                    throw new DataStreamInitException("找不到bean名称为:" + nextBeanName + "的Stream");
                }

                nextComp = (IComponent) beansWithAnnotation.get(nextBeanName);
                nextComp.setName(nextStreamName);
                DataStream ds = AnnotationUtils.findAnnotation(nextComp.getClass(), DataStream.class);
                log.info("data-stream {} 下游关系 {} --> {}", currStreamType, currentStreamName, nextStreamName);
                manager.addComponent(ds.type(), nextComp);
                collect.add(nextComp);
            }
            currentComp.addOutputStream(collect);
        }
        manager.init();
        log.info("data-stream 注解加载完成");
        return manager;
    }

    private DataStreamManager loadOutStreamConfig(DataStreamManager manager) throws DataStreamInitException {
        log.info("data-stream 加载xml配置");
        DataStreamConfigInfo dataStreamConfig = streamConfig.getDataStreamConfig();
        Map<String, ComponentNode> nodeMap = dataStreamConfig.getNodeMap();

        List<ComponentNode> outputStreamList = dataStreamConfig.getOutputStreamList();
        addToMap(nodeMap, StreamType.OUTPUT, outputStreamList);
        log.debug("output stream:{}", outputStreamList.toString());

        List<ComponentNode> streamList = dataStreamConfig.getStreamList();
        addToMap(nodeMap, StreamType.STREAM, streamList);
        log.debug("stream:{}", streamList.toString());

        List<ComponentNode> inputStreamList = dataStreamConfig.getInputStreamList();
        addToMap(nodeMap, StreamType.INPUT, inputStreamList);
        log.debug("input stream:{}", inputStreamList.toString());

        initStream(manager, StreamType.OUTPUT, outputStreamList, nodeMap);
        initStream(manager, StreamType.STREAM, streamList, nodeMap);
        initStream(manager, StreamType.INPUT, inputStreamList, nodeMap);
        log.info("data-stream xml加载完成");
        return manager;
    }

    private void addToMap(Map<String, ComponentNode> nodeMap, StreamType type, List<ComponentNode> nodeList) {
        if (nodeList == null) {
            return;
        }

        for (ComponentNode node : nodeList) {
            node.setType(type);
            nodeMap.put(node.getName(), node);
            log.debug("data-stream 配置加载到map {} {}", type, node.getName(), node.getBeanName());
        }
    }

    private void initStream(DataStreamManager manager, StreamType type, List<ComponentNode> componentNodeList,
        Map<String, ComponentNode> nodeMap) throws DataStreamInitException {
        if (componentNodeList == null) {
            return;
        }
        for (ComponentNode node : componentNodeList) {
            if (node == null) {
                continue;
            }
            String nodeName = node.getName();
            log.info("data-stream初始化{} componentName:{} beanName:{}", type, nodeName, node.getBeanName());

            IComponent component = initStream(manager, type, node);
            component.init();
            String outStreamStr = config.getProperty(nodeName + ".outputs");

            Set<String> outputs =node.getOutputs()==null? Sets.newLinkedHashSet(): Sets.newLinkedHashSet(node.getOutputs());
            if(StringUtils.isNotBlank(outStreamStr)){
                String[] split = outStreamStr.split(",");
                log.info("加载配置文件中 {}.outputs={}",nodeName,outStreamStr);
                Collections.addAll(outputs,split);
            }
            if (outputs != null) {
                for (String out : outputs) {
                    IComponent outStream = manager.getComponent(out);
                    if (outStream == null) {
                        log.debug("data-stream {} {} 下游 {} 没有初始化，开始初始化", type, nodeName, out);
                        ComponentNode outNode = nodeMap.get(out);
                        if (outNode == null) {
                            log.error("data-stream 配置错误，找不到{},请检查配置", out);
                            throw new IllegalArgumentException(out);
                        }
                        outStream = initStream(manager, outNode.getType(), outNode);
                    }
                    if (outStream != null) {
                        log.info("data-stream {} 下游关系 {} --> {}", type, nodeName, outStream.getName());
                        component.addOutputStream(outStream);
                    }
                }
            }
        }
    }

    private IComponent initStream(DataStreamManager manager, StreamType type, ComponentNode node) {
        if (node == null) {
            return null;
        }
        log.info("initStream BeanName {} Name {}", node.getBeanName(), node.getName());
        if (manager.containsComponent(node.getName())) {
            IComponent component = manager.getComponent(node.getName());
            log.info("manager存在 Name {} 直接返回{}", component.getName(), component);
            return component;
        }

        log.info("从spring获取 BeanName {} Name {}", node.getBeanName(), node.getName());
        IComponent component = context.getBean(node.getBeanName(), IComponent.class);
        component.setName(node.getName());
        log.debug("data-stream {} 获取bean:{} 设置stream名:{}/{} class:{}", type, node.getBeanName(), node.getName(), component.getName(),
            component);

        manager.addComponent(type, component);
        return component;
    }

}
